/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.dubbo.rpc.cluster.loadbalance;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.support.RpcUtils;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Round robin load balance.
 */
public class RoundRobinLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "roundrobin";

    private static final int RECYCLE_PERIOD = 60000;
    /**
     * key：服务接口名称
     * value： key = 提供者ip+ 服务接口名称 。value= 轮询权重实例对
     */
    private final ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>();

    protected static class WeightedRoundRobin {
        // 服务提供者权重值
        private int weight;

        // 轮询权重的值
        private final AtomicLong current = new AtomicLong(0);

        // 最后更新时间
        private long lastUpdate;

        public int getWeight() {
            return weight;
        }

        public void setWeight(int weight) {
            this.weight = weight;
            current.set(0);
        }

        public long increaseCurrent() {
            // 加权值
            return current.addAndGet(weight);
        }

        public void sel(int total) {
            current.addAndGet(-1 * total);
        }

        public long getLastUpdate() {
            return lastUpdate;
        }

        public void setLastUpdate(long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }

    /**
     * get invoker addr list cached for specified invocation
     * <p>
     * <b>for unit test only</b>
     *
     * @param invokers
     * @param invocation
     * @return
     */
    protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + RpcUtils.getMethodName(invocation);
        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
        if (map != null) {
            return map.keySet();
        }
        return null;
    }

    /**
     *
     * 加权轮询，按公约后的权重设置轮询比率，循环调用节点
     * 缺点：同样存在慢的提供者累积请求的问题。
     *
     * @param invokers
     * @param url
     * @param invocation
     * @return
     * @param <T>
     */
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // 获取服务接口名：com.huan.dubbo.service.HelloService.sayHello
        String key = invokers.get(0).getUrl().getServiceKey() + "." + RpcUtils.getMethodName(invocation);
        /**
         * methodWeightMap
         * key：服务接口名称
         * value： key = 提供者ip+ 服务接口名称 。value= 轮询权重实例对
         */
        ConcurrentMap<String, WeightedRoundRobin> map = ConcurrentHashMapUtils.computeIfAbsent(methodWeightMap, key, k -> new ConcurrentHashMap<>());

        int totalWeight = 0;
        // 记录轮询最大的权重
        long maxCurrent = Long.MIN_VALUE;

        /**
         * 案例，现在有3个服务提供者：
         *      服务提供者（权重值）：A（3）  B（1）  C（5）
         *      权重总和值 totalWeight： 3+1+5 =9
         *
         * 代码逻辑：
         *
         *   服务提供者（权重值,轮询权重初始值0）：   A（3,0）        B（1,0）       C（5,0）
         *              第1次 rpc 请求：   A（3,0+3=3）    B（1,0+1=1）   C（5,0+5=5）
         *                   本次调用：轮询权重最大是C，然后修改 C服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+5 =-4
         *              第2次 rpc 请求：   A（3,3+3=6）    B（1,1+1=2）   C（5,-4+5=1）
         *                   本次调用：轮询权重最大是A，然后修改 A服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+6 =-3
         *              第3次 rpc 请求：   A（3,-3+3=0）    B（1,2+1=3）   C（5,1+5=6）
         *                   本次调用：轮询权重最大是C，然后修改 C 服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+6 =-3
         *              第4次 rpc 请求：   A（3,0+3=3）    B（1,3+1=4）   C（5,-3+5=2）
         *                   本次调用：轮询权重最大是B，然后修改 B服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+4 =-5
         *              第5次 rpc 请求：   A（3,3+3=6）    B（1,-5+1=-4）   C（5,2+5=7）
         *                   本次调用：轮询权重最大是C，然后修改 C服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+7 =-2
         *              第5次 rpc 请求：   A（3,6+3=9）    B（1,-4+1=-3）   C（5,-2+5=3）
         *                   本次调用：轮询权重最大是A，然后修改 A服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+9 =0
         *              ....
         *
         *   上面就是代码执行逻辑，要进记住！！！
         */
        long now = System.currentTimeMillis();
        // 记录最终负载均衡算法得出的 Invoker
        Invoker<T> selectedInvoker = null;
        // 记录最终负载均衡算法得出的 轮询权重
        WeightedRoundRobin selectedWRR = null;

        // 查找最大轮询权重值的  Invoker，也是加权轮询算法核心！！
        for (Invoker<T> invoker : invokers) {
            // 获取提供者主机+服务接口名 如：dubbo://192.168.31.130:30882/com.huan.dubbo.service.HelloService
            String identifyString = invoker.getUrl().toIdentityString();

            // 获取该 invoker 权重
            int weight = getWeight(invoker, invocation);

            // 创建轮询权重对象 （该对象是轮询的关键所在）
            WeightedRoundRobin weightedRoundRobin = ConcurrentHashMapUtils.computeIfAbsent(map, identifyString, k -> {

                WeightedRoundRobin wrr = new WeightedRoundRobin();
                // 设置当前轮询权重 值
                wrr.setWeight(weight);
                return wrr;
            });


            /**
             * 如果权重不相同证明这个提供者被修改了权重并重启。在没有修改的时候可能有其它消费者来使用这个提供者了，并且使用是轮询加权负载均衡算法；
             * 当消费者再次使用轮询加权负载均衡算法这个提供者的时候。发现这个提供者的权重和本地缓存中的权重不一致了，
             * 那么就要更新本地内存，把本地缓存的权重设置为初始化；然后再重新 轮询加权负载均衡
             */
            if (weight != weightedRoundRobin.getWeight()) {
                //weight changed
                // 更新当前轮询权重 值
                weightedRoundRobin.setWeight(weight);
            }

            // 修改轮询权重值，为其增加服务提供者权重值
            long cur = weightedRoundRobin.increaseCurrent();

            // 设置最后更新时间
            weightedRoundRobin.setLastUpdate(now);

            // 判断 当前服务提供者轮询权重值 是否大于 当前最大轮询权重值
            if (cur > maxCurrent) {

                // 记录当前服务提供者轮询权重值 为最大轮询权重值
                maxCurrent = cur;
                // 记录当前服务提供者为本次轮询胜出者
                selectedInvoker = invoker;
                // 记录当前服务提供者的 轮询权重对象
                selectedWRR = weightedRoundRobin;

            }
            // 权重相加 得到权重总和
            totalWeight += weight;
        }


        /**
         * 判断提供者数量 不等于 消费者本地缓存提供者数量
         * true：证明 有些服务提供者已经下线了
         *
         */
        if (invokers.size() != map.size()) {
            // 移除超过十分钟没有更新的 服务提供者轮询权重对象
            map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
        }


        if (selectedInvoker != null) {
            /**
             * 修改轮询权重值。轮询权重值 - 服务提供者权重总和值，为了使这个服务提供者轮询权重值变为最小轮询权重值
             */
            selectedWRR.sel(totalWeight);

            return selectedInvoker;
        }


        // should not happen here
        /**
         *  代码不可能执行到这里的
         */
        return invokers.get(0);
    }

}
